package utils;

// Establishes the connection to one mymed node and retrieves
// information about the ring. 
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.management.MBeanServerConnection;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;

import org.apache.cassandra.dht.Token;

import cassandra.ColumnFamilyThreadPool;
import cassandra.Ring;
import exception.RuleBasedMonitoringException;

public class CassandraJMXProxy {

	Ring ring;
	MBeanServerConnection mbsc;	
	ObjectName storageService;
	public final String storageServiceName = "StorageService";
	public final String threadPoolExecutorName = "JMXEnabledThreadPoolExecutor";
	static CassandraJMXProxy singleton;

	public static CassandraJMXProxy getInstance () throws RuleBasedMonitoringException {
		if (singleton == null) {
			singleton = new CassandraJMXProxy();
		}
		return singleton;
	}

	public CassandraJMXProxy () throws RuleBasedMonitoringException {
		InetAddress ip;
		String entryPointNode = ConfigProperties.getInstance().getEntryPoint();
		try {

			ip = InetAddress.getByName(entryPointNode);
			storageService = getObjectName(ip, storageServiceName);
		} catch (UnknownHostException e) {

			throw (new RuleBasedMonitoringException("Failed to get the node with name: " + entryPointNode));
		} 		

	}




	public ObjectName getObjectName (InetAddress Ip, String objectNameName) throws RuleBasedMonitoringException {

		int Portnumber = 8080;
		ObjectName name = null;
		Set mbeans = null;


		try {

			JMXConnector jmxc = Utils.remoteConnection(Ip, Portnumber);
			mbsc = Utils.mbeanServer(jmxc);

			Set names = Utils.mbeanNames(mbsc);

			for (Iterator i = names.iterator(); i.hasNext();) {
				name = (ObjectName) i.next();

				if (name.toString().contains("org.apache.cassandra")) {
					mbeans = Utils.serverMbeans(mbsc, name);
					for (Iterator j = mbeans.iterator(); j.hasNext();) {
						if (j.next().toString().contains(objectNameName))
							return name;
					}
				}
			}			

		} catch (Exception e) {
			throw (new RuleBasedMonitoringException("Failed to connect to node with ID " + Ip.getHostName(), e));
		}

		throw (new RuleBasedMonitoringException("Failed to obtain storage service of node " + Ip.getHostName()));
	}

	public Ring getRingImage () throws RuleBasedMonitoringException {
		try {
			ring = new Ring((Map<String, String>)mbsc.getAttribute(storageService,"LoadMap"), (List<String>)mbsc.getAttribute(storageService,"LiveNodes"),  (List<String>) mbsc.getAttribute(storageService,"UnreachableNodes"), (Map<Token, String>) mbsc.getAttribute(storageService, "TokenToEndpointMap"));
			//Object [] params = {null};
			//String [] signature = {String.class.getName()};


		} catch (Exception e) {
			throw (new RuleBasedMonitoringException("Failed to get ring image", e));
		}     	 
		return ring;
	}

	public List<ColumnFamilyThreadPool> getColumnThreadPools () throws RuleBasedMonitoringException {
		List<ColumnFamilyThreadPool> list = new ArrayList<ColumnFamilyThreadPool>(); 
		Map<String, String> ipNodes = ring.getLoadMap();
		for (String ip : ipNodes.keySet()) {
			try {

				ObjectName node   =	CassandraJMXProxy.getInstance().getObjectName(InetAddress.getByName(ip), threadPoolExecutorName);
				long pendingTask = (Long) mbsc.getAttribute(node, "PendingTasks");
				int activeCount = (Integer) mbsc.getAttribute(node, "ActiveCount");				
				ColumnFamilyThreadPool pool = new ColumnFamilyThreadPool(node.getCanonicalName(), ip, activeCount, pendingTask);
				list.add(pool);
			} catch (Exception e) {
				throw (new RuleBasedMonitoringException("Failed to get column family thread pools", e));
			}     	 
		}

		return list;
	}

	public void callOperation (ObjectName object, String operationName, Object[] params, String[] signature) throws RuleBasedMonitoringException {
		try {
			mbsc.invoke(object, operationName, params, signature);
		} catch (Exception e) {
			throw (new RuleBasedMonitoringException(e));
		}     	 
	}


}
